package car;

import Utils.JSONDeserializationSchema;
import Utils.JdbcUtils;
import com.yege.bean.Student;
import com.yege.sink.HBaseSinkFunction;
import com.yege.sink.HbaseBaseMap;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.formats.json.JsonNodeDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Properties;

public class Test4_lllegalVehiclesTrack {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.getProperty("bootstrap.servers", "hadoop10:9092");
        properties.getProperty("group.id", "g3");

        /*FlinkKafkaConsumer<MonitorInfo> consumer = new FlinkKafkaConsumer<MonitorInfo>("topic-car", new JSONDeserializationSchema<>(MonitorInfo.class), properties);
        DataStreamSource<MonitorInfo> ds1 = env.addSource(consumer);


        ds1.filter(new RichFilterFunction<MonitorInfo>() {

            Connection connection;
            PreparedStatement ps;
            ResultSet rs;

            @Override
            public void open(Configuration parameters) throws Exception {
                connection = JdbcUtils.getconnection();
                ps = connection.prepareStatement("select speed_limit from t_monitor_info where monitor_id = ?");
            }

            @Override
            public boolean filter(MonitorInfo value) throws Exception {
                ps.setString(1,value.getCar());
                rs = ps.executeQuery();
                
                boolean flag = false;
                if (rs.next()){
                    flag = true;
                }
                return flag;
            }

            @Override
            public void close() throws Exception {
                JdbcUtils.release(rs,ps,connection);
            }
        }).map(new HbaseBaseMap<>())
                .addSink(new HBaseSinkFunction("t_track_info","cf1",1));*/

        env.execute();

    }
}
